package org.lwd.datalink.core.probe;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.hyperic.sigar.SigarException;
import org.lwd.datalink.api.probe.WorkerSystemStateProbe;
import org.lwd.datalink.core.runtime.WorkerConfig;
import org.lwd.datalink.domain.worker.WorkerInfo;
import org.lwd.datalink.domain.worker.WorkerSystemStateInfo;
import org.lwd.datalink.service.SysPropertiesService;
import org.lwd.datalink.service.WorkerService;
import org.lwd.datalink.service.WorkerSystemStateService;
import org.lwd.datalink.system.SystemSnapshot;
import org.lwd.datalink.system.SystemUtils;
import org.lwd.datalink.utils.NamedThreadFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.MessageFormat;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * worker机器系统状态统计
 *
 * @Author: weidong.lin
 * @Date: create in 2024/4/9 14:58
 */
@Slf4j
@Service
public class WorkerSystemStateProbeImpl implements WorkerSystemStateProbe {

    private volatile boolean running = false;

    private long lastIncomingNetworkTraffic;//上一时刻接收的总字节数

    private long lastOutgoingNetworkTraffic;//上一时刻发送的总字节数

    private ScheduledExecutorService workerSystemStateExecutor;

    @Autowired
    private WorkerService workerService;

    @Autowired
    private WorkerSystemStateService workerSystemStateService;

    @Autowired
    private SysPropertiesService sysPropertiesService;

    @Override
    public void start() {
        if (running) {
            return;
        }

        long period = 1000 * 60;
        workerSystemStateExecutor = Executors.newScheduledThreadPool(1,
                new NamedThreadFactory(MessageFormat.format("Probe-Type-{0}", WorkerSystemStateProbe.class.getSimpleName())));
        workerSystemStateExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    getWorkerSystemState();
                } catch (Exception e) {
                    log.error("Worker system state probe failed.", e);
                }
            }
        }, period, period, TimeUnit.MILLISECONDS);

        log.info("Worker system state probe started.");
        running = true;
    }

    @Override
    public void stop() {
        if (!running) {
            return;
        }
        workerSystemStateExecutor.shutdownNow();
        log.info("Worker system state probe stopped.");
    }

    private void getWorkerSystemState() throws SigarException {
        // TODO 收集HotSpot
//        SystemSnapshot systemSnapshot = SystemUtils.buildSystemSnapshot();
//        WorkerSystemStateInfo systemStateInfo = new WorkerSystemStateInfo();
//        systemStateInfo.setLoadAverage(systemSnapshot.getLoadAverage());
//        systemStateInfo.setUserCPUUtilization(systemSnapshot.getUserCPUUtilization());
//        systemStateInfo.setSysCPUUtilization(systemSnapshot.getSysCPUUtilization());
//        Map<String, Long> incomingMap = systemSnapshot.getIncomingNetworkTrafficMap();
//        Map<String, Long> outgoingMap = systemSnapshot.getOutgoingNetworkTrafficMap();
//        Map<String,String> sysMap = sysPropertiesService.map();
//        String netTrafficName = StringUtils.isNotBlank(sysMap.get("net_traffic_name")) ? sysMap.get("net_traffic_name") : "eth0";
//        long incoming = incomingMap.get(netTrafficName);
//        long incomingNetworkTraffic = incoming - lastIncomingNetworkTraffic;
//        lastIncomingNetworkTraffic = incoming;
//        long outgoing = outgoingMap.get(netTrafficName);
//        long outgoingNetworkTraffic = outgoing - lastOutgoingNetworkTraffic;
//        lastOutgoingNetworkTraffic = outgoing;
//        systemStateInfo.setIncomingNetworkTraffic(incomingNetworkTraffic);
//        systemStateInfo.setOutgoingNetworkTraffic(outgoingNetworkTraffic);
//        systemStateInfo.setTcpCurrentEstab(systemSnapshot.getTcpCurrentEstab());
//        String clientId = WorkerConfig.current().getString(CommonClientConfigs.CLIENT_ID_CONFIG);
//        systemStateInfo.setWorkerId(Long.valueOf(clientId));
//        WorkerInfo workerInfo = workerService.getById(Long.valueOf(clientId));
//        if (workerInfo != null) {
//            systemStateInfo.setHost(workerInfo.getWorkerAddress());
//        }
//
//        workerSystemStateService.insert(systemStateInfo);
    }
}
